package com.hownoon.parquet;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.example.GroupReadSupport;

import java.io.IOException;
import java.util.ArrayList;
import java.util.UUID;

public class ReadDataThread implements Runnable {

    Connection connection;
    String path;
    String id;
    ArrayList<Put> puts=new ArrayList<>();
    int sum=0;

    public ReadDataThread(Connection connection,String path,String id) {
        this.connection=connection;
        this.path=path;
        this.id=id;
    }

    @Override
    public void run() {

        System.out.println("id<====>"+id);
        GroupReadSupport readSupport = new GroupReadSupport();
        ParquetReader.Builder<Group> reader= ParquetReader.builder(readSupport, new Path(path));
        ParquetReader<Group> build= null;

        System.out.println("读取开始"+path);
        long startTime = System.currentTimeMillis();
        try {
            build = reader.build();
            Group line=null;

            /*while((line=build.read())!=null){
                *//*int ask_seq_num=line.getInteger("ask_seq_num", 0);
                int bid_seq_num=line.getInteger("bid_seq_num", 0);
                String bs_flag=line.getString("bs_flag", 0);
                String date=line.getString("date", 0);
                String instrument=line.getString("instrument", 0);
                float price=line.getFloat("price", 0);
                int seq_num=line.getInteger("seq_num", 0);
                int volume=line.getInteger("volume", 0);
                long __index_level_0__=line.getLong("__index_level_0__", 0);*//*

                String bs_flag=line.getString("bs_flag", 0);
                String date=line.getString("date", 0);
                date=date.replace(" ","&");
                String price=String.valueOf(line.getFloat("price", 0));
                String volume=String.valueOf(line.getInteger("volume", 0));

                if(bs_flag.equals("B")){
                    Calendar calendar = Calendar.getInstance();
                    calendar.setTime(new SimpleDateFormat("yyyy-MM-dd&HH:mm:ss.SSS").parse(date));
                    String part;
                    long pid=(calendar.getTimeInMillis()/10)%100;
                    if(pid<10){
                        part="0"+pid;
                    }else{
                        part=Long.toString(pid);
                    }
                    String uuid = UUID.randomUUID().toString().replaceAll("-","");
                    String rowKey=part+"_"+date+"_"+uuid;

                    Put put = new Put(Bytes.toBytes(rowKey));
                    put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("price"), Bytes.toBytes(price));
                    put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("volume"), Bytes.toBytes(volume));
                    put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("date"), Bytes.toBytes(date));
                    //put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("bs_flag"), Bytes.toBytes(bs_flag));
                    puts.add(put);

                }
                if(puts.size()==1000){
                    Table table = connection.getTable(TableName.valueOf("trade"));
                    table.put(puts);
                    table.close();
                    puts.clear();
                    System.out.println("线程<===>"+id+"<===>"+"插入条数<===>"+sum+"/"+(sum+1000));
                    sum=sum+1000;

                }

                *//*System.out.println(
                        line.getInteger("ask_seq_num", 0)
                                +"\t"+line.getInteger("bid_seq_num", 0)
                                +"\t"+line.getString("bs_flag", 0)
                                +"\t"+line.getString("date", 0)
                                +"\t"+line.getString("instrument", 0)
                                +"\t"+line.getFloat("price", 0)
                                +"\t"+line.getInteger("seq_num", 0)
                                +"\t"+line.getInteger("volume", 0)
                                +"\t"+line.getLong("__index_level_0__", 0)
                );*//*
            }*/

            for(int i=0;i<1000000;i++){
                String part;
                long pid=((i+1)/10)%100;
                if(pid<10){
                    part="0"+pid;
                }else{
                    part=Long.toString(pid);
                }
                String date="2018-06-01&23:23.233";
                String price="23";
                String volume="2308";
                String uuid = UUID.randomUUID().toString().replaceAll("-","");
                String rowKey=part+"_"+date+"_"+uuid;

                Put put = new Put(Bytes.toBytes(rowKey));
                put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("price"), Bytes.toBytes(price));
                put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("volume"), Bytes.toBytes(volume));
                put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("date"), Bytes.toBytes(date));
                //put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("bs_flag"), Bytes.toBytes(bs_flag));
                puts.add(put);

                if(puts.size()==1000){
                    Table table = connection.getTable(TableName.valueOf("trade"));
                    table.put(puts);
                    table.close();
                    puts.clear();
                    System.out.println("线程<===>"+id+"<===>"+"插入条数<===>"+sum+"/"+(sum+1000));
                    sum=sum+1000;

                }
            }



           /* Table table = connection.getTable(TableName.valueOf("trade"));
            sum=sum+puts.size();
            System.out.println("线程<===>"+id+"<===>"+"插入条数<===>"+sum+"/"+sum);
            table.put(puts);
            table.close();
            puts.clear();*/


        } catch (IOException e) {
            e.printStackTrace();
        } /*catch (ParseException e) {
            e.printStackTrace();
        }*/

        long endTime = System.currentTimeMillis();    //获取结束时间

        System.out.println("******************************************************************************");
        System.out.println("线程"+id+"插入完毕<==>文件路径<==>"+path+"<==>插入总条数"+sum+"<==>线程耗时"+(endTime - startTime) + "ms");
        System.out.println("******************************************************************************");

    }



}
